package fun.ticsmyc.multiThread;

import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.SocketException;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;


/**
 * UDP服务器
 *
 * @author Ticsmyc
 * @package fun.ticsmyc.multiThread
 * @date 2020-03-12 11:37
 */
public class Server {
    private static final int PORT=18888;
    private static final int MAX_LENGTH=1024;

    private DatagramSocket datagramSocket;

    /**
     * 记录响应次数，线程安全用原子变量保证
     */
    private AtomicInteger modCount;

    private LinkedBlockingQueue<DatagramPacket> receiveQueue;
    private LinkedBlockingQueue<DatagramPacket> responseQueue;

    public Server(){
        try {
            datagramSocket = new DatagramSocket(PORT);
            receiveQueue =  new LinkedBlockingQueue<DatagramPacket>();
            responseQueue =  new LinkedBlockingQueue<DatagramPacket>();
            modCount = new AtomicInteger(0);
        } catch (SocketException e) {
            e.printStackTrace();
        }
    }

    /**
     * 接收线程，监听端口，收到数据包后直接放到接收队列
     */
    private void startReceive(){
        System.out.println("接收开启");
        //三个接收线程
        for(int i=0;i<3;++i){
            new Thread(()->{
                while(true){
                    //阻塞接收， 收到后就放入队列
                    byte[] buffer =new byte[MAX_LENGTH];
                    DatagramPacket receivePacket = new DatagramPacket(buffer,buffer.length);
                    try {
                        datagramSocket.receive(receivePacket);
                        receiveQueue.put(receivePacket);
                    } catch (IOException | InterruptedException e) {
                        e.printStackTrace();
                    }
                }

            },"receiveThread").start();
        }
    }

    /**
     * 处理线程，从接收队列取数据包，计算后封装新的包放入发送队列
     */
    private void startProcess(){
        System.out.println("处理开启");
        //开3个处理线程
        for(int i=0; i<3 ;++i){
            new Thread(()->{
                while(true){
                    DatagramPacket receivePacket=null;
                    try {
                        //阻塞式取数据
                        receivePacket = receiveQueue.take();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    String parameter  = new String(receivePacket.getData(),0,receivePacket.getLength());
                    String[] opers = parameter.split("\n");
                    double num1 = Double.parseDouble(opers[1]);
                    double num2 = Double.parseDouble(opers[2]);
                    //System.out.println(num1 +" "+opers[0]+" "+num2);
                    double res =0;
                    switch(opers[0]){
                        case "+" :
                            res = num1+num2;
                            break;
                        case "-":
                            res=num1-num2;
                            break;
                        case "x":
                            res=num1*num2;
                            break;
                        case "/":
                            res=num1/num2;
                            break;
                    }

                    //在原来的数据包上修改一下，省得再new一个
                    receivePacket.setData((res+"").getBytes());
                    try {
                        //放入相应队列
                        responseQueue.put(receivePacket);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }

                }
            },"processThread").start();
        }
    }

    /**
     * 从发送队列取数据包，发出去
     */
    private void startResponse(){
        System.out.println("响应开启");
        //开三个发送线程
        for(int i=0;i<3;++i){
            new Thread(()->{
                while(true){
                    try {
                        DatagramPacket responsePacket = responseQueue.take();
                        datagramSocket.send(responsePacket);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }

                    System.out.println("响应: "+ modCount.addAndGet(1));
                }
            },"responseThread").start();
        }
    }

    /**
     * 不知道在哪里关socket了，，就放这里吧 垃圾回收的时候关233333
     */
    protected void finalize(){
        if(datagramSocket!=null){
            datagramSocket.close();
        }
    }

    public static void main(String[] args) {
        Server server = new Server();

        server.startResponse();
        server.startProcess();
        server.startReceive();

    }
}
